#ifndef REDIS_CONNECTOR_H
#define REDIS_CONNECTOR_H

#include <muduo/net/EventLoop.h>
#include <muduo/net/Channel.h>
#include <muduo/base/Logging.h>
#include <hiredis.h>
#include <async.h>
#include <boost/noncopyable.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/bind.hpp>

using namespace boost;
using namespace muduo::net;

namespace sdfs
{
namespace redis
{

typedef boost::function<void (redisReply*)> redisCmdCallback;
//typedef void (redisCmdCallback)(redisReply* );

void connectCallback(const redisAsyncContext *c, int status) {
    if (status != REDIS_OK) {
        LOG_ERROR << "can not connect REDIS, error code: " << c->errstr;
        return;
    }

    LOG_TRACE << "Connected to REDIS successfully";
}

void disconnectCallback(const redisAsyncContext *c, int status) {
    if (status != REDIS_OK) {
        LOG_ERROR << "can not disconnect REDIS, error code: " << c->errstr;
        return;
    }
	LOG_TRACE << "Disconnected to REDIS successfully";
}


void enableRead(void *privdata)
{
	Channel* channel_ = reinterpret_cast<Channel*>(privdata);
	channel_->enableReading();
}

void disableRead(void *privdata)
{
	//Channel* channel_ = reinterpret_cast<Channel*>(privdata);
}

void enableWrite(void *privdata)
{
	Channel* channel_ = reinterpret_cast<Channel*>(privdata);
	channel_->enableWriting();
}

void disableWrite(void *privdata)
{
	Channel* channel_ = reinterpret_cast<Channel*>(privdata);
	channel_->disableWriting();
}

void disableAll(void *privdata)
{
	Channel* channel_ = reinterpret_cast<Channel*>(privdata);
	channel_->disableAll();
}

void cmdCallback(redisAsyncContext *c, void *r, void *privdata) {
    redisReply *reply = reinterpret_cast<redisReply*>(r);
    if (reply == NULL) return;
    redisCmdCallback* cb = reinterpret_cast<redisCmdCallback*>(privdata);
    (*cb)(reply);
}
class AsyncRedisconnector : public boost::noncopyable
{
public:
	AsyncRedisconnector(EventLoop *loop, const char *ip, int port):
		loop_(loop)
	{
		ac_ = redisAsyncConnect(ip, port);
		assert( ac_ != NULL );
		channel_.reset(new Channel(loop, ac_->c.fd));
		
		channel_->enableReading();
		channel_->setReadCallback(
			boost::bind(&AsyncRedisconnector::readCallback, this, _1));
		channel_->setWriteCallback(
			boost::bind(&AsyncRedisconnector::writeCallback, this));

		attachToHiredis(loop, ac_);

		redisAsyncSetConnectCallback(ac_,connectCallback);
		redisAsyncSetDisconnectCallback(ac_,disconnectCallback);
		//channel_->set_index(-1);
		//loop_->updateChannel(channel_.get());

	}
	~AsyncRedisconnector()
	{
		redisAsyncDisconnect(ac_);
		loop_->removeChannel(channel_.get());
	}

	int cmd(redisCmdCallback& fn, const char* format, ...)
	{
		void* pf = reinterpret_cast<void*>(&fn);
		va_list ap;
		int status;
		va_start(ap,format);
		status = redisvAsyncCommand(ac_,cmdCallback,pf,format,ap);
		va_end(ap);
		return status;
	}

private:

	void attachToHiredis(EventLoop* loop, redisAsyncContext* c)
	{
		c->ev.addRead = enableRead;
		c->ev.delRead = disableRead;
		c->ev.addWrite = enableWrite;
		c->ev.delWrite = disableWrite;
		c->ev.cleanup = disableAll;

		c->ev.data = channel_.get();
	}

	void readCallback( muduo::Timestamp timestamp)
	{
		redisAsyncHandleRead(ac_);
	}

	void writeCallback()
	{
		redisAsyncHandleWrite(ac_);
	}

	/* data */
private:
	EventLoop* loop_;
	scoped_ptr<Channel> channel_;
	redisAsyncContext* ac_;
	redisCmdCallback cmdCallback_;
};

}
}

#endif